package com.feng.cloud.common.producer;

import com.alibaba.fastjson.JSON;
import com.feng.cloud.common.constants.RocketMqConstant;
import com.feng.cloud.common.message.DelayMessage;
import com.feng.cloud.common.utils.DelayLevelCalculate;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.remoting.exception.RemotingException;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.MessagingException;

import java.time.Duration;
import java.time.Instant;
import java.util.Optional;

/**
 * @ClassName FengDelayProducer
 * @Description TODO
 * @Author Better Feng
 * @Date 2021/1/25 16:57
 **/
@Slf4j
public class FengDelayProducer {

    @Autowired
    private RocketMQTemplate rocketMQTemplate;

    public SendResult sendDelayMessage(DelayMessage delayMessage) throws MessagingException, InterruptedException, RemotingException, MQClientException, MQBrokerException {
        Optional.ofNullable(delayMessage.getStartDeliverDate()).orElseThrow(() -> new IllegalArgumentException("定时消息不能为空"));
        // 获取当前距离定时时间的差
        long duration = Duration.between(Instant.now(), delayMessage.getStartDeliverDate().toInstant()).getSeconds();
        // 获取延时等级
        int level = DelayLevelCalculate.calculateDefault(duration);
        Message message = new Message(delayMessage.getTopic(),delayMessage.getTags(),delayMessage.getBody());
        if (level > RocketMqConstant.MINI_DELAY_LEVEL) {
            // 如果大于最小延时等级，设置延时发送
            message.setDelayTimeLevel(level);
            message.setTopic(RocketMqConstant.DEFAULT_DELAY_TOPIC);
            message.setBody(JSON.toJSONString(delayMessage).getBytes());
        }
        return rocketMQTemplate.getProducer().send(message);
    }
}
